package com.sugon.ohdfs.integration.flink.job.stream.source;

import com.sugon.ohdfs.integration.flink.domain.TestItem;
import com.sugon.ohdfs.integration.flink.service.ConsoleSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.PostConstruct;
import java.io.Serializable;

public abstract class AbstractStreamSourceJob implements Serializable{

    @Value("${flink.checkpoint.interval:60000}")
    long checkPointInterval;
    @Value("${hdfs.checkpoint.path:hdfs://10.11.8.29:9001/flink/sinkTest/checkpoint}")
    String checkPointFilePath;
    @Value("${hdfs.savepoint.path:hdfs://10.11.8.29:9001/flink/sinkTest/savepoint}")
    String savePointFilePath;

    protected abstract FileSource buildFileSource();
    protected abstract TestItem transform(Object item);


    @PostConstruct
    public void init() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //IMPORTANT: Checkpointing needs to be enabled when using the FileSink in STREAMING mode. Part files can only
        // be finalized on successful checkpoints. If checkpointing is disabled, part files will forever stay in the
        // in-progress or the pending state, and cannot be safely read by downstream systems.
        env.enableCheckpointing(checkPointInterval, CheckpointingMode.EXACTLY_ONCE);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage(checkPointFilePath);

        FileSource fileSource = buildFileSource();
        DataStreamSource dataStreamSource = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "");
        dataStreamSource
                .map(item -> transform(item))
                .filter(item -> item!=null)
                .addSink(new ConsoleSink());
        env.execute(getClass().getSimpleName());
    }

}
